package com.ft.turorial.spring.boot.kafka.producer;

import com.alibaba.fastjson.JSON;
import com.ft.turorial.spring.boot.kafka.message.MessageWrapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * \* Created with IntelliJ IDEA.
 * \* User: Administrator
 * \* Date: 2016/11/30
 * \* Time: 10:53
 * \* To change this template use File | Settings | File Templates.
 * \* Description:
 * \
 */
public class Sender {
    private static final Logger LOGGER = LoggerFactory
            .getLogger(Sender.class);

    @Autowired
    private KafkaTemplate<Integer, String> kafkaTemplate;

    public void sendMessage(String topic, MessageWrapper messageWrapper) {

        // the KafkaTemplate provides asynchronous send methods returning a
        String message  = JSON.toJSONString(messageWrapper);
        // Future
        ListenableFuture<SendResult<Integer, String>> future = kafkaTemplate
                .send(topic, message);

        // you can register a callback with the listener to receive the result
        // of the send asynchronously
        future.addCallback(
                new ListenableFutureCallback<SendResult<Integer, String>>() {

                    @Override
                    public void onSuccess(SendResult<Integer, String> result) {
                        LOGGER.info("sent message='{}' with offset={}",
                                message,
                                result.getRecordMetadata().offset());
                    }

                    @Override
                    public void onFailure(Throwable ex) {
                        LOGGER.error("unable to send message='{}'",
                                message, ex);
                    }
                });

        // alternatively, to block the sending thread, to await the result,
        // invoke the future’s get() method
    }
}
